package com.shujia.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel

/**
 * Kryo序列化是spark提供的一种专门适配spark计算时提高性能的一种序列化方式
 *
 * 序列化可以提高计算的性能，加快计算速度，减少数据所占用的内存空间
 */

case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

object Demo7Kryo {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      //将序列化方式设置为Kryo的序列化方式
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      //自定义一个序列化类，指定要序列化的东西
      .config("spark.kryo.registrator", "com.shujia.opt.Demo8KryoRegistrator")
      .getOrCreate()
    val sparkContext: SparkContext = sparkSession.sparkContext

    val studentsRDD: RDD[Student] = sparkContext.textFile("spark/data/students.csv").map((line: String) => {
      val infos: Array[String] = line.split(",")
      Student(infos(0), infos(1), infos(2).toInt, infos(3), infos(4))
    })


    /**
     * 未做序列化：238.3 KiB
     * 使用默认的序列化方式：55.7 KiB
     * 使用kryo序列化：43.0 KiB
     *
     */
    //针对被复用的rdd进行缓存
    //    studentsRDD.cache() // 默认是MEMORY_ONLY
    //    studentsRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)
//    studentsRDD.persist(StorageLevel.MEMORY_ONLY)
    studentsRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    /**
     * 统计每个班级的人数
     */
    studentsRDD.map((stu: Student) => (stu.clazz, 1))
      .reduceByKey(_ + _)
      .saveAsTextFile("spark/data/opt/clazz_num")

    /**
     * 统计每个性别的人数
     *
     * 第一次作用用到studentsRDD的时候是原本的数据量大小，当第二个作业也用到studentsRDD数据的时候，就去缓存中寻找数据
     */
    studentsRDD.map((stu: Student) => (stu.gender, 1))
      .reduceByKey(_ + _)
      .saveAsTextFile("spark/data/opt/gender_num")

    while (true) {

    }
  }

}
